package com.jonah.demo.hbase.thrift;

import com.jonah.demo.hbase.thrift.common.JsonUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;


public class HbaseServiceImpl implements HbaseService {
    private static final Logger LOG = LoggerFactory.getLogger(HbaseServiceImpl.class);
    private final String encoding = "utf-8";
    private Configuration conf;

    public Configuration getConf() {
        return conf;
    }

    public void setConf(Configuration conf) {
        this.conf = conf;
    }

    @Override
    public Map<String, Object> getRow(String tableName, List<Map<String, String>> columns, String rowKey) throws IOException {
        Map<String, Object> map = new HashMap<>();
        try (Connection connection = ConnectionFactory.createConnection(conf);
             Table hTable = connection.getTable(TableName.valueOf(tableName))) {
            Get get = new Get(Bytes.toBytes(rowKey));
            Result result = hTable.get(get);
            for (Map<String, String> column : columns) {
                for (Map.Entry<String, String> entry : column.entrySet()) {
                    String family = entry.getKey();
                    String qualifier = entry.getValue();
                    String value = Bytes.toString(result.getValue(Bytes.toBytes(family), Bytes.toBytes(qualifier)));
                    map.put(qualifier, value);
                }

            }
        }
        return map;
    }


    @Override
    public List<Map<String, Object>> getRows(String tableName, List<Map<String, String>> columns, List<Get> rowKey) throws IOException {
//        LOG.debug("HbaseServiceImpl.getRows.input[tableName->{}]",tableName);
//        LOG.debug("HbaseServiceImpl.getRows.input[columns->{}]",JsonUtil.toJson(columns));
//        LOG.debug("HbaseServiceImpl.getRows.input[rowKey->{}]",JsonUtil.toJson(rowKey));
        List<Map<String, Object>> resultMap = new ArrayList<>();
        try (Connection connection = ConnectionFactory.createConnection(conf);
             Table hTable = connection.getTable(TableName.valueOf(tableName))) {
            Result[] results = hTable.get(rowKey);
            LOG.info("HbaseServiceImpl.getRows.results.size->{}", results.length);
            for (Result result : results) {
                String row = Bytes.toString(result.getRow());
                LOG.debug("rowKey = >{}", row);
                if (row == null) {
                    continue;
                }
                Map<String, Object> map = new HashMap<>();
                map.put("rowKey", row);
                for (Map<String, String> column : columns) {
                    for (Map.Entry<String, String> entry : column.entrySet()) {
                        String family = entry.getKey();
                        String qualifier = entry.getValue();
                        String value = Bytes.toString(result.getValue(Bytes.toBytes(family), Bytes.toBytes(qualifier)));
                        map.put(qualifier, value);
                    }
                }
                resultMap.add(map);
                LOG.debug("map = >{}", map);
            }
        }
        return resultMap;
    }

    @Override
    public int addData(String rowKey, String tableName, String family, String[] column, String[] value) throws IOException {
        assert column!=null&&value!=null&&column.length==value.length;
        int result = 0;
        try (Connection connection = ConnectionFactory.createConnection(conf);
            Table hTable = connection.getTable(TableName.valueOf(tableName))) {
            Put put = new Put(Bytes.toBytes(rowKey));
            for (int i = 0; i < column.length; i++) {
                put.addColumn(Bytes.toBytes(family),Bytes.toBytes(column[i]),Bytes.toBytes(value[i]));
                result ++;
            }
            hTable.put(put);
        }
        return result;
    }



    @Override
    public <PARAM, RESULT> List<RESULT> getRows(String tableName, FilterList filterList, HBaseServiceAfter<PARAM, RESULT> after) throws Exception {
        LOG.debug("HbaseServiceImpl.getRows.input[tableName->{}]", tableName);
        LOG.debug("HbaseServiceImpl.getRows.input[filterList->{}]", filterList);
        List<RESULT> results = new ArrayList<>();
        try (Connection connection = ConnectionFactory.createConnection(conf);
             Table hTable = connection.getTable(TableName.valueOf(tableName))) {
            Scan scan = new Scan();
            scan.setFilter(filterList);
            scan.setMaxResultSize(10);
//            scan.setLimit(10);
            scan.setMaxVersions();
            ResultScanner scanner = hTable.getScanner(scan);

            for (Result result : scanner) {
                if (result.isEmpty()) {
                    continue;
                }
                LOG.debug("getRows. hbase scan result", JsonUtils.toJson(result));
                RESULT after1 = after.after((PARAM) result);
                if(after1!=null){
                    results.add(after1);
                }

            }
        }
        LOG.debug("HbaseServiceImpl.getRows.input[filterList.result.size->{}]", results.size());
        return results;
    }

    @Override
    public <PARAM, RESULT> List<RESULT> getRows(String tableName, List<Get> filterList, HBaseServiceAfter<PARAM, RESULT> after) throws Exception {
        LOG.debug("HbaseServiceImpl.getRows.input[tableName->{}]", tableName);
        LOG.debug("HbaseServiceImpl.getRows.input[filterList->{}]", filterList);
        List<RESULT> results = new ArrayList<>();
        try (Connection connection = ConnectionFactory.createConnection(conf);
             Table hTable = connection.getTable(TableName.valueOf(tableName))) {
            Result[] resultGets = hTable.get(filterList);


            for (Result result : resultGets) {
                if (result.isEmpty()) {
                    continue;
                }
                LOG.debug("getRows. hbase get result", JsonUtils.toJson(result));
                RESULT after1 = after.after((PARAM) result);
                if(after1!=null){
                    results.add(after1);
                }

            }
        }
        LOG.debug("HbaseServiceImpl.getRows.input[filterList.result.size->{}]", results.size());
        return results;
    }

    @Override
    public Boolean puts(String tableName, List<Put> puts) throws Exception {
        LOG.debug("HbaseServiceImpl.getRows.input[tableName->{}]", tableName);
        try (Connection connection = ConnectionFactory.createConnection(conf);
             Table hTable = connection.getTable(TableName.valueOf(tableName))) {
            hTable.put(puts);

        }
        return Boolean.TRUE;
    }

    @Override
    public Boolean puts(String tableName, Put put) throws Exception {
        LOG.debug("HbaseServiceImpl.getRows.input[tableName->{}]", tableName);
        try (Connection connection = ConnectionFactory.createConnection(conf);
             Table hTable = connection.getTable(TableName.valueOf(tableName))) {
            hTable.put(put);

        }
        return Boolean.TRUE;
    }

}
